package com.itheima.flinkmomo.flink;


import com.itheima.flinkmomo.pojo.MoMoCountBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;


public class MysqlSink extends RichSinkFunction<MoMoCountBean> {
    private Statement stat;
    private Connection connection;

    //private String sql;
    private String status;

    public MysqlSink() {
    }

    public MysqlSink(String status) {
        this.status = status;
    }

    /**
     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();

        stat = connection.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭连接和释放资源
        if (connection != null) {
            connection.close();
        }
        if (stat != null) {
            stat.close();
        }
    }

    /**
     * 每条数据的插入都要调用一次 invoke() 方法
     *
     * @param value
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(MoMoCountBean value, Context context) throws Exception {

        if(status.equals("1")){
            String sql = "select * from momo_count where momo_groupType = '1'";
            ResultSet resultSet = stat.executeQuery(sql);
            boolean flag = resultSet.next();

            if(flag) {
                sql = "update momo_count set momo_totalcount= '"+value.getMoMoTotalCount()+ "' where momo_groupType = '1'";
            }else {
                sql = "insert into momo_count( momo_totalcount,momo_groupType) values ("+value.getMoMoTotalCount()+",'1') ";
            }
            stat.executeUpdate(sql);


        }else if (status.equals("2")){
            String sql = "select * from momo_count where momo_groupType = '2' and momo_province= '"+value.getMoMoProvince()+"' ";
            ResultSet resultSet = stat.executeQuery(sql);
            boolean flag = resultSet.next();

            if(flag) {
                sql = "update momo_count set momo_msgcount= '"+value.getMoMo_MsgCount()+ "' where momo_groupType = '2' and momo_province= '"+value.getMoMoProvince()+"' ";
            }else {
                sql = "insert into momo_count( momo_province,momo_msgcount,momo_groupType) values ('"+value.getMoMoProvince()+"',"+value.getMoMo_MsgCount()+",'2') ";
            }
            stat.executeUpdate(sql);
        }else if (status.equals("3")){
            String sql = "select * from momo_count where momo_groupType = '3' and momo_province= '"+value.getMoMoProvince()+"' ";
            ResultSet resultSet = stat.executeQuery(sql);
            boolean flag = resultSet.next();

            if(flag) {
                sql = "update momo_count set momo_msgcount= '"+value.getMoMo_MsgCount()+ "' where momo_groupType = '3' and momo_province= '"+value.getMoMoProvince()+"' ";
            }else {
                sql = "insert into momo_count( momo_province,momo_msgcount,momo_groupType) values ('"+value.getMoMoProvince()+"',"+value.getMoMo_MsgCount()+",'3') ";
            }
            stat.executeUpdate(sql);

        }else if (status.equals("4")){
            String sql = "select * from momo_count where momo_groupType = '4' and momo_username= '"+value.getMoMoUsername()+"' ";
            ResultSet resultSet = stat.executeQuery(sql);
            boolean flag = resultSet.next();

            if(flag) {
                sql = "update momo_count set momo_msgcount= '"+value.getMoMo_MsgCount()+ "' where momo_groupType = '4' and momo_username= '"+value.getMoMoUsername()+"' ";
            }else {
                sql = "insert into momo_count( momo_username,momo_msgcount,momo_groupType) values ('"+value.getMoMoUsername()+"',"+value.getMoMo_MsgCount()+",'4') ";
            }
            stat.executeUpdate(sql);


        }else if (status.equals("5")){

            String sql = "select * from momo_count where momo_groupType = '5' and momo_username= '"+value.getMoMoUsername()+"' ";
            ResultSet resultSet = stat.executeQuery(sql);
            boolean flag = resultSet.next();

            if(flag) {
                sql = "update momo_count set momo_msgcount= '"+value.getMoMo_MsgCount()+ "' where momo_groupType = '5' and momo_username= '"+value.getMoMoUsername()+"' ";
            }else {
                sql = "insert into momo_count( momo_username,momo_msgcount,momo_groupType) values ('"+value.getMoMoUsername()+"',"+value.getMoMo_MsgCount()+",'5') ";
            }
            stat.executeUpdate(sql);

        }

    }

    private static Connection getConnection() {
        Connection con = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            con = DriverManager.getConnection("jdbc:mysql://node1:3306/momo?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());
        }
        return con;
    }
}
